package com.gitee;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerTest {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers","node01:9092");
        props.setProperty("group.id","ke"); //当前消费者组id
        props.setProperty("enable.auto.commit","true"); //自动提交偏移量
        props.setProperty("auto.commit.interval.ms","1000"); //每隔1000毫秒提交一次偏移量
        /*
         **earliest**
         * 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费

         **latest**
         * 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据

         **none**
         * topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常

         **latest 这个设置容易丢失消息**，假如kafka出现问题，还有数据往topic中写，这个时候重启kafka，这个设置会从最新的offset开始消费,中间出问题的哪些就不管了。*/
        props.setProperty("auto.offset.reset","earliest"); //当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
        props.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");//key反序列化
        props.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");//value反序列化
        //创建消费者对象,配置消费者参数配置
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("study_kafka3"));
        while(true){
            //阻塞获取数据,等待100毫秒
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
    }
}
